package com.atguigu.edu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.common.EduConfig;
import com.atguigu.edu.util.DruidPhoenixDSUtil;
import com.atguigu.edu.util.JedisUtil;
import com.atguigu.edu.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
    private DruidDataSource dataSource;


    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        //upsert into t (col1,col2,col3) values(v1,v2,v3)
        DruidPooledConnection connection = null;
//        Jedis jedis = null;
        try{
            connection = dataSource.getConnection();
//            jedis = JedisUtil.getJedis();
            String sinkTable = jsonObject.getString("sink_table");

            // 如果类型为修改删除redis中的旁路缓存
//            String type = jsonObject.getString("type");
//            if ("update".equals(type) || "delete".equals(type)){
//                jedis.del(sinkTable.toUpperCase() + jsonObject.getString("id"));
//            }
//            jedis.close();

            jsonObject.remove("sink_table");
//            jsonObject.remove("type");
            StringBuilder stringBuilder = new StringBuilder();

            Set<String> strings = jsonObject.keySet();
            Collection<Object> values = jsonObject.values();

            stringBuilder.append("upsert into ")
                    .append(EduConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append(" (")
                    .append(StringUtils.join(strings,","))
                    .append(") values( '")
                    .append(StringUtils.join(values,"','"))
                    .append("')");

            System.out.println(stringBuilder.toString());
            PhoenixUtil.sqlExecute(connection,stringBuilder.toString());
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("写入数据到phoenix错误");
        }finally {
            if (connection!=null){
                try{
                    connection.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
}
